package com.atguigu.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Map;
import java.util.Set;

/*
Dim App 8.自定义sink，将过滤后的数据内容写出到phoenix的表内容里
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {//hbaseDS.addSink，输入的就是hbaseDS的类型JSONObject

    /*maxwell格式：
    这里的value是主流过滤了部分字段后的value，另外这个value还增加了sinkTable
{"database":"gmall","table":"cart_info","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,"data":{"id":100924,"user_id":"93","sku_id":16,"cart_price":4488,"sku_num":1,"img_url":"http://47.93.148.192:8080/group1/M0rBHu8l-sklaALrngAAHGDqdpFtU741.jpg","sku_name":"华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万30倍数字变焦 8GB+128GB亮黑色全网通5G手机","is_checked":null,"create_time":"2020-06-14 09:28:57","operate_time":null,"is_ordered":1,"order_time":"2021-10-17 09:28:58","source_type":"2401","source_id":null},"old":{"is_ordered":0,"order_time":null}}

{
    "database":"gmall","table":"cart_info","type":"update","ts":1592270938,"xid":13090,"xoffset":1573,
    "data":{//过滤后的字段
    "id":100924,"user_id":"93","sku_id":16,"cart_price":4488,"sku_num":1
    },
    "old":{
        "is_ordered":0,"order_time":null
    }
    sink_table:"dim_xxx"
}
     */
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //todo 1.获取连接（每条数据都获取一个连接，然后归还给连接池）
        DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();


        //todo 2.拼接sql语句:upsert into db.tablename(id,name,sex) values('1001','zhangsan','male')//其实增删改都是upsert，本案例不需要删
        JSONObject data = value.getJSONObject("data");
        String sink_table = value.getString("sink_table");
        String phoenixUpsertSql= genUpsertSql(sink_table,data);//主流根据配置流过滤出来value
        System.out.println("SQL"+phoenixUpsertSql);

        //TODO 如果为更新数据，则先将数据写到redis（旁路缓存，当维表有更新的时候需要添加这一步）
        if ("update".equals(value.getString("type"))){
            Jedis jedis = JedisUtil.getJedis();
//            String redisKey = "DIM:" + tableName + ":" + key;
            String redisKey = "DIM:" + sink_table.toUpperCase() + ":" + data.getString("id");

            //TODO 将data数据转为大写 写到redis
            JSONObject jsonObject = new JSONObject();
            Set<Map.Entry<String, Object>> entries = data.entrySet();
            for (Map.Entry<String, Object> entry : entries) {
                jsonObject.put(entry.getKey().toUpperCase(),entry.getValue().toString());//entry.getValue().toString()要变成toString，phoenix查询后缓存到redis的列值是带双引号的，如果没有toString，更新的时候写入redis就没有带双引号了，就不一致了
            }
            jedis.set(redisKey,jsonObject.toJSONString());
            jedis.expire(redisKey,24*3600);


            jedis.close();

        }

        //todo 3.再将数据写到phoenix
        PreparedStatement preparedStatement = phoenixConn.prepareStatement(phoenixUpsertSql);
        preparedStatement.execute();
        phoenixConn.commit();//DML(增删改查)操作需要手动、DDL(建表)操作会自动提交************************！！！

        //todo 4.释放资源
        preparedStatement.close();
        phoenixConn.close();

    }

    //todo 自定义phoenix的插入sql语句
//    upsert into db.tablename(id,name,sex) values('1001','zhangsan','male')//其实增删改都是upsert，本案例不需要删
    private String genUpsertSql(String sinkTable, JSONObject data) {//把JSONObject的方法当作hashmap处理
        //取出列名[id name sex]=>"id,name,sex"
        Set<String> columns = data.keySet();
        //取出列值
        Collection<Object> values = data.values();

        return "upsert into "+ GmallConfig.PHOENIX_DB+"."+sinkTable+"("+
                StringUtils.join(columns,",")+//StringUtils(lang3)会自动把set集合的columns自动转化为字符串
                ")values('"+
                StringUtils.join(values,"','")+
                "')"
                ;
    }


}
